package com.hzgj.bcl.soa.command;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import com.hzgj.bcl.soa.config.ServiceDef;
import com.hzgj.bcl.soa.monitor.MonitorService;
import com.hzgj.bcl.soa.registry.Consumer;
import com.hzgj.bcl.soa.registry.Provider;
import com.hzgj.bcl.util.reg.EtcdExtClient;

import lombok.Getter;
import lombok.Setter;
import mousio.etcd4j.promises.EtcdResponsePromise;
import mousio.etcd4j.responses.EtcdException;
import mousio.etcd4j.responses.EtcdKeysResponse;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import static com.hwtx.soa.registry.EtcdConstants.*;
import static com.hzgj.bcl.soa.registry.RegistryConstants.SERVICE_DOWN;
import static com.hzgj.bcl.soa.registry.RegistryConstants.SERVICE_PREFIX;

/**
 * Created by Administrator on 2016/8/11.
 */
@Component
public class EtcdClient {

    private static int timeout = 5;

    protected static Logger logger = LoggerFactory.getLogger(EtcdClient.class);
    protected static EtcdExtClient client;
    private static Properties properties = new Properties();

    static {
        InputStream inputStream = Thread.currentThread().getContextClassLoader().getResourceAsStream("app.properties");
        try {
            properties.load(inputStream);
        } catch (IOException e) {
            logger.error("{}", e);
        }
        init();
    }

    public static void main(String[] args) throws Exception {
        while (true) {
            EtcdResponsePromise<EtcdKeysResponse> promise = client.waitForKey("/service/com.hzgj.soa.service.UserService");
            EtcdKeysResponse response = promise.get();
            if (response.getNode().key.contains("com.hzgj.soa.service.UserService")) {
                System.out.println(response.action);
                System.out.println(response.getNode().getValue());
            }
        }
    }

    private static void init() {
        String addr = properties.getProperty("etcd.address");
        List<URI> uris = Lists.newArrayList();
        for (String url : addr.split(",")) {
            if (!url.startsWith("http://")) {
                url = "http://" + url;
            }
            uris.add(URI.create(url));
        }

        try {
            client = new EtcdExtClient(timeout, TimeUnit.SECONDS, uris);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public List<Consumer> getConsumes(String service, String address) {
        try {
            String path = REFERENCE_PREFIX + "/" + service + "/" + SERVICE_CONSUMES + "/" + address;
            EtcdKeysResponse.EtcdNode node = client.getDirValue(path);
            return getClients(node);
        } catch (EtcdException e) {
            if (e.errorCode != 100) {
                logger.error("{}", e);
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
        return Collections.emptyList();
    }

    public List<Consumer> getConsumes(String service) {
        final List<Consumer> clients = Lists.newArrayList();
        try {
            String path = REFERENCE_PREFIX + "/" + service + "/" + SERVICE_CONSUMES + "/";
            List<EtcdKeysResponse.EtcdNode> nodes = client.getDirWithRecursive(path);
            if (nodes != null) {
                nodes.stream().forEach(etcdNode -> {
                    clients.addAll(getClients(etcdNode));
                });
            }
        } catch (EtcdException e) {
            if (e.errorCode != 100) {
                logger.error("{}", e);
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
        return clients;
    }

    private List<Consumer> getClients(EtcdKeysResponse.EtcdNode etcdNode) {
        if (etcdNode != null) {
            try {
                Object object = JSON.parse(etcdNode.value);
                if (object instanceof JSONObject) {
                    return Lists.newArrayList(JSON.parseObject(etcdNode.value, Consumer.class));
                } else {
                    return JSON.parseArray(etcdNode.value, Consumer.class);
                }
            } catch (Exception e) {
                logger.error("{}", e);
            }
        }
        return Collections.EMPTY_LIST;
    }

    /**
     * serviceName:clientInfo
     */
    public Map<String, ClientInfo> getConsumes() {
        final Map<String, ClientInfo> clients = Maps.newHashMap();
        try {
            List<EtcdKeysResponse.EtcdNode> nodes = client.getDir(REFERENCE_PREFIX);
            if (nodes != null && nodes.size() > 0) {
                nodes.stream().forEach(node -> {
                    String key = node.key;
                    ClientInfo clientInfo = new ClientInfo();
                    try {
                        clientInfo.setCount(getClientCount(key));
                        clientInfo.setService(key.substring(REFERENCE_PREFIX.length() + 1));
                        clients.putIfAbsent(clientInfo.getService(), clientInfo);
                    } catch (EtcdException e) {
                        if (e.errorCode != 100) {
                            logger.error("{}", e);
                        }
                    } catch (Exception e) {
                        logger.error("{}", e);
                    }
                });
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
        return clients;
    }

    @Setter
    @Getter
    public static class ClientInfo {
        String service;
        int count;
    }

    interface BuildServiceCallback {
        int getConsumes(String service);
    }

    public Map<String, ServiceDef.ServiceInfo> getServices(BuildServiceCallback buildServiceCallback) {
        final Map<String, ServiceDef.ServiceInfo> services = Maps.newHashMap();
        try {
            List<EtcdKeysResponse.EtcdNode> nodes = client.getDir(SERVICE_PREFIX);
            if (nodes != null) {
                nodes.stream().forEach(node -> {
                    try {
                        if (!node.key.endsWith(MonitorService.class.getName())) {
                            List<EtcdKeysResponse.EtcdNode> _nodes = client.getDirWithRecursive(node.key);
                            if (_nodes != null && _nodes.size() > 0) {
                                ServiceDef.ServiceInfo serviceInfo = getServiceInfo(_nodes);
                                serviceInfo.setName(node.key.substring(SERVICE_PREFIX.length() + 1));
                                if (buildServiceCallback != null) {
                                    serviceInfo.setConsumes(buildServiceCallback.getConsumes(serviceInfo.getName()));
                                }
                                services.putIfAbsent(serviceInfo.getName(), serviceInfo);
                            }
                        }
                    } catch (EtcdException e) {
                        if (e.errorCode != 100) {
                            logger.error("{}", e);
                        }
                    } catch (Exception e) {
                        logger.error("{}", e);
                    }
                });
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
        return services;
    }


    private ServiceDef.ServiceInfo getServiceInfo(List<EtcdKeysResponse.EtcdNode> nodes) {
        ServiceDef.ServiceInfo serviceInfo = new ServiceDef.ServiceInfo();
        nodes.forEach(node -> {
            String key = node.key;
            if (key.endsWith(SERVICE_INFO)) {
                ServiceDef.ServiceInfo _info = JSON.parseObject(node.value, ServiceDef.ServiceInfo.class);
                serviceInfo.setComment(_info.getComment());
            } else if (key.endsWith(SERVICE_PROVIDERS)) {
                serviceInfo.setProviders(node.nodes.size());
            }
        });
        return serviceInfo;
    }

    public Map<String, Provider> getProviders(String name) {

        Map<String, Provider> providers = Maps.newHashMap();
        try {
            buildProvider(client.getDirWithRecursive(getServicePrefixPath(name)), providers, true);
            buildProvider(client.getDirWithRecursive(getDownServicePath(name)), providers, false);
        } catch (EtcdException e1) {
            if (e1.errorCode != 100) {
                logger.error("{}", e1);
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
        return providers;
    }

    private int getProvidersCount(String key) throws Exception {
        int up_size = 0;
        try {
            List<EtcdKeysResponse.EtcdNode> nodes = client.getDir(key + "/" + SERVICE_PROVIDERS);
            if (nodes != null) {
                up_size += nodes.size();
            }
            nodes = client.getDir(key + "/" + SERVICE_DOWN);
            if (nodes != null) {
                up_size += nodes.size();
            }
        } catch (EtcdException e1) {
            if (e1.errorCode != 100) {
                logger.error("{}", e1);
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
        return up_size;
    }

    private int getClientCount(String key) throws Exception {
        int size = 0;
        try {
            List<EtcdKeysResponse.EtcdNode> nodes = client.getDir(key + "/" + SERVICE_CONSUMES);
            if (nodes != null) {
                size += nodes.size();
            }
        } catch (EtcdException e1) {
            if (e1.errorCode != 100) {
                logger.error("{}", e1);
            }
        } catch (Exception e) {
            logger.error("{}", e);
        }
        return size;
    }

    public void remService(String service) throws Exception {
        int count = getProvidersCount(SERVICE_PREFIX + "/" + service);
        if (count > 0) {
            throw new Exception("can't delete service which contain providers.");
        }
        client.deleteDirWithRecursive(SERVICE_PREFIX + "/" + service);
    }

    public void remClient(String service) throws Exception {
        int count = getClientCount(REFERENCE_PREFIX + "/" + service);
        if (count > 0) {
            throw new Exception("can't delete client which contain consumes.");
        }
        client.deleteDirWithRecursive(REFERENCE_PREFIX + "/" + service);
    }

    void buildProvider(List<EtcdKeysResponse.EtcdNode> nodes, Map<String, Provider> providers, boolean up) {
        if (nodes != null) {
            nodes.forEach(etcdNode -> {
                Object object = JSON.parse(etcdNode.value);
                if (object instanceof JSONObject) {
                    Provider provider = JSON.parseObject(etcdNode.value, Provider.class);
                    provider.setUp(up);
                    providers.put(provider.getAddress(), provider);
                } else {
                    for (Provider provider : JSON.parseArray(etcdNode.value, Provider.class)) {
                        providers.put(provider.getAddress(), provider);
                    }
                }
            });
        }
    }
}
